package com.sdy.hospitalguard.web.websocket.base;

import com.sdy.hospitalguard.biz.constants.RedisConstants;
import com.sdy.hospitalguard.biz.event.WebsocketEventCallback;
import com.sdy.hospitalguard.biz.event.WebsocketEvent;
import com.sdy.mvc.utils.JsonUtil;
import com.sdy.redis.service.RedisService;
import lombok.Data;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;

import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
public abstract class BaseWebsocketServer {
    private static final int USER_MAX_CONNECTION_COUNT = 10000;
    protected static ConcurrentHashMap<String, BaseWebsocketServer> webSocketMap = new ConcurrentHashMap<>();
    protected Session session;
    protected String connectionId;
    protected Integer userId;
    private RedisService redisService;
    
    protected abstract ApplicationContext getApplicationContext();
    
    public static String createConnectionId(RedisService redisService, Integer userId) {
        if (userId == null) {
            return null;
        }
        String connId = UUID.randomUUID().toString().replace("-", "");
        redisService.set(RedisConstants.WEBSOCKET_CONNECTION_USER + connId, userId, 600);
        return connId;
    }

    @OnOpen
    public void onOpen(Session session, @PathParam("connectionId") String connectionId) throws IOException {
        this.redisService = getApplicationContext().getBean(RedisService.class);
        this.session = session;
        this.connectionId = connectionId;
        this.userId = redisService.get(RedisConstants.WEBSOCKET_CONNECTION_USER + this.connectionId, Integer.class);
        if (this.userId == null) {
            this.session.close();
            return;
        }
        webSocketMap.put(this.connectionId, this);
        // 单个用户连接数量过多，清空吧
        if (redisService.scard(RedisConstants.WEBSOCKET_USER_CONNECTION + userId) > USER_MAX_CONNECTION_COUNT) {
            redisService.del(RedisConstants.WEBSOCKET_USER_CONNECTION + userId);
        }
        redisService.sadd(RedisConstants.WEBSOCKET_USER_CONNECTION + this.userId, this.connectionId);
        log.info("[连接ID:{}] 建立连接, 当前连接数:{}", this.connectionId, webSocketMap.size());
    }

    @OnClose
    public void onClose() {
        webSocketMap.remove(this.connectionId);
        redisService.sremove(RedisConstants.WEBSOCKET_USER_CONNECTION + this.userId, this.connectionId);
        log.info("[连接ID:{}] 断开连接, 当前连接数:{}", connectionId, webSocketMap.size());
    }

    @OnError
    public void onError(Session session, Throwable error) {
        log.error("[连接ID:{}] 连接错误", this.connectionId, error);
    }

    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("[连接ID:{}] 收到消息:{}", this.connectionId, message);
    }

    public synchronized boolean sendMessage(String message) {
        try {
            this.session.getBasicRemote().sendText(message);
            return true;
        } catch (IOException e) {
            log.error("[连接ID:{}] 发送消息失败, 消息:{}", this.connectionId, message, e);
        }
        return false;
    }
    
    @Data
    @Accessors(chain = true)
    public static class WebsocketResponse {
        private String eventId;
        private String eventType;
        private Object data;
        private String toJson() {
            return JsonUtil.toJson(this);
        }
    }

    @Component
    @Slf4j
    public static class WebsocketEventListener implements ApplicationListener<WebsocketEvent> {
        @Autowired
        private RedisService redisService;
        @Autowired
        private ApplicationContext applicationContext;
        @Override
        public void onApplicationEvent(WebsocketEvent event) {
            log.info("消费事件ID={}, TYPE={}, USER_ID={}", event.getEventId(), event.getEventType(), event.getUserId());
            // 不处理超过30分钟的消息
            if (Math.abs(event.getTimestamp() - System.currentTimeMillis()) > 30 * 60 * 1000L) {
                log.warn("消费事件ID={} 事件超时，拒绝消费", event.getEventId());
                return;
            }
            AtomicBoolean success = new AtomicBoolean(false);
            Set<Object> userConns = redisService.smember(RedisConstants.WEBSOCKET_USER_CONNECTION + event.getUserId());
            userConns.forEach(connId -> {
                BaseWebsocketServer websocketServer = webSocketMap.get(connId.toString());
                if (websocketServer == null) {
                    log.debug("[连接ID:{}] 发送消息失败，客户端已断开连接。", connId);
                    return;
                }
                String resp = new WebsocketResponse()
                        .setEventId(event.getEventId())
                        .setEventType(event.getEventType())
                        .setData(event.getData())
                        .toJson();
                if (websocketServer.sendMessage(resp)) {
                    success.set(true);
                }
            });
            if (!success.get()) {
                log.warn("发送消息失败，没有活跃的客户端。EventId={}, UserId={}", event.getEventId(), event.getUserId());
                if (event.getCallback() != null) {
//                    int queryTimes = 10;
//                    while (!isEventSuccess(event) && queryTimes-- > 0) {
//                        try {
//                            Thread.sleep(3000);
//                        } catch (InterruptedException e) {
//                            log.error("", e);
//                        }
//                    }
                    if (!isEventSuccess(event)) {
                        if (setEventFailed(event)) {
                            Class<? extends WebsocketEventCallback> cb = event.getCallback();
                            try {
                                Method method = cb.getMethod("onError", ApplicationContext.class, WebsocketEvent.class);
                                method.invoke(cb.newInstance(), applicationContext, event);
                            } catch (NoSuchMethodException | InvocationTargetException | InstantiationException | IllegalAccessException e) {
                                log.error("", e);
                            }
                        }
                    }
                }
            } else {
                log.info("发送消息成功。EventId={}, Type={}, UserId={}", event.getEventId(), event.getEventType(), event.getUserId());
                if (setEventSuccess(event)) {
                    if (event.getCallback() != null) {
                        Class<? extends WebsocketEventCallback> cb = event.getCallback();
                        try {
                            Method method = cb.getMethod("onSuccess", ApplicationContext.class, WebsocketEvent.class);
                            method.invoke(cb.newInstance(), applicationContext, event);
                        } catch (NoSuchMethodException | InvocationTargetException | InstantiationException | IllegalAccessException e) {
                            log.error("", e);
                        }
                    }
                }
            }
        }
        
        private boolean isEventSuccess(WebsocketEvent event) {
            return redisService.exists(RedisConstants.WEBSOCKET_EVENT_SUCCESS_FLAG + event.getEventId());
        }

        private boolean setEventSuccess(WebsocketEvent event) {
            return redisService.setnx(RedisConstants.WEBSOCKET_EVENT_SUCCESS_FLAG + event.getEventId(), 1, 3600);
        }
        
        private boolean setEventFailed(WebsocketEvent event) {
            return redisService.setnx(RedisConstants.WEBSOCKET_EVENT_FAILED_FLAG + event.getEventId(), 1, 3600);
        }
    }
}
